package com.vhall.component.service.room.listener;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.vhall.component.dao.room.RoomExtendsMapper;
import com.vhall.component.entity.room.dto.RoomsEntityDTO;
import com.vhall.component.entity.room.entity.RoomExtendsEntity;
import com.vhall.component.plugin.core.service.PaasService;
import com.vhall.component.plugin.core.service.dto.BigDataBroadcastRecordReportDTO;
import com.vhall.component.service.room.RoomInnerService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;

import java.util.Objects;

/**
 * Redis 队列消息订阅
 * 大数据-房间直播数据上报
 *
 * @author yuanzh
 * @date 2021/11/26
 */
@Slf4j
@Lazy
@Component
public class BigDataBroadcastRecordReportStreamListener implements StreamListener<String, ObjectRecord<String, String>> {

    @Value("${vhall.bigdata.broadcast-record-group-name:bigdata-broadcast-record-group}")
    private String groupName;

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private RoomExtendsMapper roomExtendsMapper;

    @Autowired
    private RoomInnerService roomInnerService;

    @Autowired
    private PaasService paasService;

    @Override
    public void onMessage(ObjectRecord<String, String> message) {
        // 消息ID，room_extends表id
        RecordId messageId = message.getId();
        // 消息的key和value
        log.info("【房间直播记录上报】接收到Bigdata Broadcast Record Stream消息 stream message. messageId={}, stream={}, body={}", messageId,
                message.getStream(), message.getValue());
        // 开始处理消息的逻辑
        String roomExtendsId = message.getValue();
        // 根据id查询对应开播记录
        LambdaQueryWrapper<RoomExtendsEntity> wrapper = Wrappers.lambdaQuery();
        wrapper.eq(RoomExtendsEntity::getId, roomExtendsId).last("limit 1");
        RoomExtendsEntity roomExtends = roomExtendsMapper.selectOne(wrapper);
        if (Objects.isNull(roomExtends)) {
            log.error("【房间直播记录上报】查询不到房间扩展信息，roomExtendsId={}", roomExtendsId);
            return;
        }
        String roomId = roomExtends.getRoomId();
        RoomsEntityDTO roomInfo = roomInnerService.getRoomInfo(roomId);
        if (Objects.isNull(roomInfo)) {
            log.error("【房间直播记录上报】查询不到房间信息，roomExtendsId={}, roomId={}", roomExtendsId, roomId);
            return;
        }

        BigDataBroadcastRecordReportDTO recordReport = new BigDataBroadcastRecordReportDTO.Builder()
                // 设置appId
                .setAppId(roomInfo.getAppId())
                // 设置活动id
                .setActivityId(String.valueOf(roomInfo.getIlId()))
                // 设置B端用户id
                .setUserId(String.valueOf(roomInfo.getAccountId()))
                // 设置房间相关id
                .setRoomIds(roomInfo.getRoomId(), roomInfo.getInavId(), roomInfo.getChannelId())
                // 设置开始时间及结束时间
                .setTime(roomExtends.getStartTime(), roomExtends.getEndTime()).build();
        // 进行数据上报
        if (paasService.broadcastRecordReport(recordReport)) {
            // 通过RedisTemplate手动确认消息
            redisTemplate.opsForStream().acknowledge(groupName, message);
            log.error("【房间直播记录上报】数据上报成功，roomExtendsId={}", roomExtendsId);
        } else {
            log.error("【房间直播记录上报】数据上报失败，roomExtendsId={}", roomExtendsId);
        }
    }
}
